package com.haisen.service.impl;

import com.alibaba.fastjson.JSON;
import com.haisen.dataobject.entity.FileQueue;
import com.haisen.dataobject.entity.TblField;
import com.haisen.dataobject.mapper.base.TblFieldMapper;
import com.haisen.service.FileQueueService;
import com.haisen.service.HiveJdbcTemplateService;
import com.haisen.service.TblFieldService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import tk.mybatis.mapper.weekend.Weekend;
import tk.mybatis.mapper.weekend.WeekendCriteria;

import java.util.*;
import java.util.stream.Collectors;

@Service
@Slf4j
public class TblFieldServiceImpl implements TblFieldService {

    @Autowired
    TblFieldMapper tblFieldMapper;
    @Autowired
    HiveJdbcTemplateService jdbcTemplateService;
    @Autowired
    private FileQueueService fileQueueService;

    String hivewarehouse = "/user/hive/warehouse";

    @Override
    public void MergeData(String targTblName,String targTblDesc, String filterTblName,List<TblField> tblFields,boolean mod) {
       String creatTableSql = InitDDL(targTblName,targTblDesc,tblFields);
       //1、获取要建表的字段名
       List<String> fields = tblFields.stream().map(o ->{
           return o.getColumnName();
       }).collect(Collectors.toList());
       //2、获取目标表中包含字段的表
        Map<String,List<String>> tblAndField = findTblAndField(filterTblName,fields,mod);
        //3、获取select from DML
        List<String> insertSqlList = GetDataSouce(targTblName,tblAndField);
        log.info("合并数据sql语句：{}",insertSqlList);
        //4、建立表格
        String retCreatSql = jdbcTemplateService.createTable(creatTableSql);
        log.info("建表结果={}",retCreatSql);
        //5、插入数据，insert into select
       for (String ins : insertSqlList){
           log.info("插入数据集：{}",ins);
          jdbcTemplateService.insertTable(ins);
       }
    }


    @Override
    public List<String> GetDataSouce(String targTblName ,Map<String,List<String>> tblAndField) {
        List<String> selectSqlList = new ArrayList<>();
        StringBuilder sb = new StringBuilder();
         for(Map.Entry<String,List<String>> entry : tblAndField.entrySet()){
             //字段Insert into table ();
              String fields  = getFields(entry.getValue());
              sb.append("insert into ");
              sb.append(targTblName);
              sb.append("(");
              sb.append(fields);
              sb.append(",cdate ");
              sb.append(")\n");
              sb.append("select ");
              sb.append(fields);
              sb.append(", '"+entry.getKey().substring(2,8)+"' as cdate ");
             sb.append("from ");
             sb.append(entry.getKey());
             selectSqlList.add(sb.toString());
             sb.delete(0,sb.length());
         }
        return selectSqlList;
    }

    private String getFields(List<String> fields){
        StringBuilder sb = new StringBuilder();
        for(String field : fields){
            sb.append(field);
            sb.append(" ,");
        }
       sb.deleteCharAt(sb.length()-1);
        return sb.toString();
    }

    @Override
    public Map<String, List<String>> findTblAndField(String filterTblName,List<String> fields,boolean mod) {
        //1、查询已导入成功过表
        List<FileQueue> fileQueueSucessList = fileQueueService.findSucess(filterTblName);
        List<String> fileQueueSucessTblNames = fileQueueSucessList.stream()
                .map(o ->{
                          return o.getTblName();
                        }).collect(Collectors.toList());
        //2、查找导入成功表的表名和字段名,类型
        Weekend<TblField> example = Weekend.of(TblField.class);
        WeekendCriteria<TblField,Object> criteria = example.weekendCriteria();
          criteria.andIn(TblField::getTblName,fileQueueSucessTblNames);
          List<TblField> tblFieldList = tblFieldMapper.selectByExample(example);
         // log.info("符合筛选条件的表格={}",JSON.toJSONString(tblFieldList));
       //查找出所有的字段名名
       // List<TblField> tblFieldList = tblFieldMapper.selectAll();

        Map<String,String> tblNameAndFiledMap = new HashMap<>();
        StringBuilder sb = new StringBuilder();
        tblFieldList.stream().collect(Collectors.groupingBy(TblField::getTblName,Collectors.toList()))
                .forEach((tblName,tblFieldListByName) ->{
                    tblFieldListByName.stream().map(o ->{
                        sb.append(o.getColumnName());
                        sb.append(",");
                        return null;
                    })
                            .collect(Collectors.toList());
                    tblNameAndFiledMap.put(tblName,sb.deleteCharAt(sb.length()-1).toString());
                    //log.info("{},{}",tblName, sb.toString());
                    sb.delete(0,sb.length());
                });
        //表格名，字段列表
         Map<String,List<String>> retMap = new HashMap<>();
         for (Map.Entry<String,String> entry : tblNameAndFiledMap.entrySet()){
             List<String> tmpList = new ArrayList<>();
             tmpList.addAll(Arrays.asList(entry.getValue().split(",")));
             tmpList.retainAll(fields);
             //1、全匹配字段，严格匹配，非贪婪模式
             if (!mod){
                 if (tmpList.size() == fields.size()){
                     retMap.put(entry.getKey(),tmpList);
                 }
             } else {
                 //2、贪婪模式，只要有字段匹配，都插入
                  if (tmpList.size() > 0) retMap.put(entry.getKey(),tmpList);
             }
         }
         log.info("包含指定字段的表格：{}",JSON.toJSONString(retMap));
        return retMap;
    }

    /**
     * 数据定义语言初始化
     * @param tableName
     * @return
     */
    private String InitDDL(String tableName,String tableDesc,List<TblField> tbFieldList){
        StringBuilder sb = new StringBuilder();
        sb.append("create table "+tableName+"(\n");
        for(TblField tblField : tbFieldList){
            sb.append(tblField.getColumnName()+" "+tblField.getTypeName()+" comment '"+tblField.getCcomment()+"',\n");
        }
        sb.append("cdate varchar(16) comment '日期'");
        if (sb.length() >1 )
            sb.deleteCharAt(sb.length()-2);
        sb.append(") COMMENT '"+tableDesc+"'\n");
        sb.append("ROW FORMAT DELIMITED\n");
        sb.append("FIELDS TERMINATED BY '|'\n");
        sb.append("COLLECTION ITEMS TERMINATED BY '-'\n");
        sb.append("MAP KEYS TERMINATED BY ':'\n");
        sb.append("LINES TERMINATED BY '\\n'");
        sb.append("\n");
       // sb.append("LOCATION 'hdfs://mycluster/user/hive"+ "/"+"hivewarehouse"+"/"+"dataset"+"/"+tableName+"'\n");
        sb.append("LOCATION 'hdfs://nameservice1"+ hivewarehouse+"/"+"dataset"+"/"+tableName+"'\n");
        return sb.toString();
    }
}
